package broker

import (
	"context"
	"encoding/json"
	"github.com/Shopify/sarama"
	"log"
)



type Options struct {
	Addrs []string
	Secure bool
	ErrorHandler Handler
	Context context.Context
}

type Option func(*Options)

type SubscribeOptions struct {
	AutoAck bool
	Queue string
	Context context.Context
}

type PublishOptions struct {
	Context context.Context
	Partitions int32
	Key string
}

type SubscribeOption func(*SubscribeOptions)

type PublishOption func(*PublishOptions)


func Addrs(addrs ...string) Option{
	return func(o *Options){
		o.Addrs = addrs
	}
}
func Partitions(p int32) PublishOption{
	return func(o *PublishOptions){
		o.Partitions = p
	}
}


type brokerConfigKey struct {}
type clusterConfigKey struct {}

var (
	DefaultBrokerConfig = sarama.NewConfig()
	DefaultClusterConfig = sarama.NewConfig()
)

func BrokerConfig(c *sarama.Config) Option{
	return setBrokerOption(brokerConfigKey{},c)
}

func ClusterConfig(c *sarama.Config) Option{
	return setBrokerOption(clusterConfigKey{},c)
}

func WithQueue(queue string) SubscribeOption{
	return func(o *SubscribeOptions){
		o.Queue = queue
	}
}

type subscribeContextKey struct{}

func SubscribeContext(ctx context.Context) SubscribeOption{
	return setSubscribeOption(subscribeContextKey{},ctx)
}

type subscribeConfigKey struct{}

func SubscribeConfig(c *sarama.Config) SubscribeOption {
	return setSubscribeOption(subscribeConfigKey{}, c)
}

type consumerGroupHandler struct {
	handler Handler
	subopts SubscribeOptions
	kopts Options
	cg sarama.ConsumerGroup
	sess sarama.ConsumerGroupSession
}

func (*consumerGroupHandler)Setup(_ sarama.ConsumerGroupSession) error{return nil}


func (*consumerGroupHandler)Cleanup(_ sarama.ConsumerGroupSession) error{return nil}


func (h *consumerGroupHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim)error{
	for msg:=range claim.Messages(){
		var m Message
		p := &publication{m: &m,t:msg.Topic,km: msg,cg:h.cg,sess: sess}
		eh := h.kopts.ErrorHandler
		if err:=json.Unmarshal(msg.Value,&m);err !=nil{
			p.err =err
			p.m.Body = msg.Value
			if eh !=nil{
				eh(p)
			}else{
				log.Fatalf("[kafka]: failed to unamarshal: %v",err)
			}
			continue
		}
		p.m.Body = msg.Value
		err := h.handler(p)
		if err ==nil && h.subopts.AutoAck{
			sess.MarkMessage(msg,"")
		}else if err!=nil{
			p.err = err
			if eh !=nil{
				eh(p)
			}else{
				log.Fatalf("[kafka]: subscriber error: %v",err)
			}
		}
	}
	return nil
}
